package com.hs.kfk.redis;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Set;

/**
 * @Description: 消费者基于redis改造
 * @Author 胡尚
 * @Date: 2024/8/8 11:19
 */
public class RedisConsumer2 {

    private final static String BOOTSTRAP_SERVER = "worker1:9092,worker2:9092,worker3:9092";
    private final static String TOPIC_NAME = "disTopic";
    private final static String CONSUMER_GROUP = "test";

    public static void main(String[] args) {
        // 设置消费者相关的属性
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 创建消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            // 消费者拉取消息
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(100));

            // 先获取所有的partition，然后每次取partition中的消息
            Set<TopicPartition> partitions = consumerRecords.partitions();
            partitions.forEach(partition -> {
                String redisKey = "redisKeyPrefex" +partition.topic() + partition.partition() + CONSUMER_GROUP;
                // TODO  根据redisKey 从redis中 获取这个消费者组下 这个partition对应的offset


                // 获取当前partition中所有的消息
                List<ConsumerRecord<String, String>> records = consumerRecords.records(partition);
                for (ConsumerRecord<String, String> record : records) {
                    long offset = record.offset();
                    // TODO 如果 redisOffset > offset 就表示重复消费了

                    // TODO 消费消息

                }
                // 获取当前partition 多个消息中的最后一个offset
                long offset = records.get(records.size() - 1).offset();
                // TODO 根据redisKey，将上方的offset存入redis中

            });

            consumer.commitSync();
        }
    }
}
